package com.redis.base.multiple;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.redis.base.redis.IRedis;
import com.redis.base.utils.JsonUtil;

import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;

public class ClusterRedis extends AbstractRedis implements IRedis{
	
	private static final Logger log = LoggerFactory.getLogger(SingleRedis.class);

	private JedisCluster jedisCluster = null;

	private RedisConfig redisConfig;

	public ClusterRedis(RedisConfig redisConfig) {
		this.redisConfig = redisConfig;
		getJedisPool();
	}

	public void getJedisPool() {
		Set<HostAndPort> nodes = new HashSet<>();
		HostAndPort hostAndPort = null;
		for(String value : redisConfig.getHostName().split(",")) {
			String[] address = value.split(":");
			hostAndPort = new HostAndPort(address[0], Integer.parseInt(address[1]));
			nodes.add(hostAndPort);
		}
		if(StringUtils.isNotEmpty(redisConfig.getPassword())) {
			jedisCluster =  new JedisCluster(nodes, redisConfig.getTimeout(), redisConfig.getTimeout(), 3, redisConfig.getPassword(), genericObjectPoolConfig());
		}else {
			jedisCluster =  new JedisCluster(nodes,genericObjectPoolConfig());
		}
	}
	
	/**
	 * redis 集群连接方案
	 * 
	 * @return
	 */
	public GenericObjectPoolConfig genericObjectPoolConfig() {
		GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
		genericObjectPoolConfig.setMaxWaitMillis(redisConfig.getMaxWaitMillis());
		genericObjectPoolConfig.setMaxTotal(redisConfig.getMaxTotal());
		genericObjectPoolConfig.setMinIdle(redisConfig.getMaxIdle());
		genericObjectPoolConfig.setMaxIdle(redisConfig.getMaxIdle());
		return genericObjectPoolConfig;
	}

	/**
	 * 获取Jedis实例
	 * 
	 * @return
	 */
	public JedisCluster getJedis() {
		try {
			return jedisCluster;
		} catch (Exception e) {
			log.error(e.getMessage());
			return null;
		}
	}

	/**
	 * 存数据到redis
	 * 
	 * @param key
	 * @param value
	 */
	@Override
	public void put(final String key, final String value) {
		put(key, value, redisConfig.getExpire());
	}

	@Override
	public void put(final String key, final String value, final Long expire) {
		JedisCluster jedis = getJedis();
		try {
			if (log.isDebugEnabled()) {
				log.debug("redis: method is put, key [" + key + "],cache_redis [" + redisConfig.isEnable() + "]");
			}
			if (!redisConfig.isEnable()) {
				return;
			}
			if (log.isDebugEnabled()) {
				log.debug("put redis====:" + key + ":" + value);
			}
			jedis.set(key, value);
			expire(jedis,key, redisConfig.getExpire());
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		}finally {
			returnResource(jedis);
		}
	}

	@Override
	public String get(final String key) {
		JedisCluster jedis = getJedis();
		try {
			if (log.isDebugEnabled()) {
				log.debug("redis: method is get, key [" + key + "],cache_redis [" + redisConfig.isEnable() + "]");
			}
			if (!redisConfig.isEnable()) {
				return "";
			}
			return (String) jedis.get(key);
		} catch (Exception e) {
			log.error(e.getMessage(), e);
			return null;
		}finally {
			returnResource(jedis);
		}
	}

	@Override
	public <T> T get(final String key, Class<T> objClass) {
		JedisCluster jedis = getJedis();
		try {
			if (log.isDebugEnabled()) {
				log.debug("redis: method is get, key [" + key + "],cache_redis [" + redisConfig.isEnable() + "]");
			}
			if (!redisConfig.isEnable()) {
				return null;
			}
			return JsonUtil.json2Bean(jedis.get(key), objClass);
		} catch (Exception e) {
			log.error(e.getMessage(), e);
			return null;
		}finally {
			returnResource(jedis);
		}
	}

	@Override
	public <T> List<T> getList(final String key, Class<T> objClass) {
		JedisCluster jedis = getJedis();
		try {
			if (log.isDebugEnabled()) {
				log.debug("redis: method is get, key [" + key + "],cache_redis [" + redisConfig.isEnable() + "]");
			}
			if (!redisConfig.isEnable()) {
				return null;
			}
			return JsonUtil.json2List(String.valueOf(jedis.get(key)), objClass);
		} catch (Exception e) {
			log.error(e.getMessage(), e);
			return null;
		}finally {
			returnResource(jedis);
		}
	}

	@Override
	public void put(final String key, final Object value) {
		put(key, value, redisConfig.getExpire());
	}

	@Override
	public void put(final String key, final Object value, final Long expire) {
		JedisCluster jedis = getJedis();
		try {
			if (log.isDebugEnabled()) {
				log.debug("redis: method is putObj, key [" + key + "],cache_redis [" + redisConfig.isEnable() + "]");
			}
			if (!redisConfig.isEnable()) {
				return;
			}
			jedis.set(key, JsonUtil.bean2Json(value));
			expire(jedis,key, redisConfig.getExpire());
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		}finally {
			returnResource(jedis);
		}
	}

	@Override
	public void unLock(final String... keys) {
		JedisCluster jedis = getJedis();
		try {
			if (log.isDebugEnabled()) {
				log.debug("redis: method is release, key [" + keys + "],cache_redis [" + redisConfig.isEnable() + "]");
			}
			if (!redisConfig.isEnable()) {
				return;
			}
			for (String key : keys) {
				String currentValueStr = String.valueOf(jedis.get(getLockKey(key)));
			    if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {
			    	jedis.del(getLockKey(key));
			    }
			}
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		}finally {
			returnResource(jedis);
		}
	}

	/**
	 * 
	 * @Title: deleteSub
	 * 
	 * @return
	 * @return: Object
	 */
	@Override
	public Object deleteLike(final String... keys) {
		JedisCluster jedis = getJedis();
		try {
			if (log.isDebugEnabled()) {
				log.debug(
						"redis: method is deleteLike, key [" + keys + "],cache_redis [" + redisConfig.isEnable() + "]");
			}
			if (!redisConfig.isEnable()) {
				return 0;
			}
			List<String> keyList = new ArrayList<String>();
			Object result = null;
			if (keys == null || keys.length <= 0) {
				return null;
			}
			for (Object key : keys) {
				Set<String> keysTemp = jedis.hkeys("*" + key + "*");
				for (String by : keysTemp) {
					result = jedis.del(by);
					keyList.add("key:" + result + ",");
				}
			}
			return result;

		} catch (Exception e) {
			log.error(e.getMessage(), e);
			return null;
		}finally {
			returnResource(jedis);
		}
	}

	@Override
	public boolean delete(final String key) {
		JedisCluster jedis = getJedis();
		try {
			if (log.isDebugEnabled()) {
				log.debug("redis: method is delete, key [" + key + "],cache_redis [" + redisConfig.isEnable() + "]");
			}
			if (!redisConfig.isEnable()) {
				return false;
			}
			jedis.del(key);
			return true;
		} catch (Exception e) {
			log.error(e.getMessage(), e);
			return false;
		}finally {
			returnResource(jedis);
		}
	}

	@Override
	public boolean tryLock(String lockKey) {
		return lock(lockKey, redisConfig.isSpin(), getSpinNum(redisConfig.getSpinNum(), DEFAULT_SPIN_NUM));
	}

	@Override
	public boolean tryLock(String lockKey, boolean isSpin) {
		return lock(lockKey, isSpin, getSpinNum(redisConfig.getSpinNum(), DEFAULT_SPIN_NUM));
	}

	@Override
	public boolean tryLock(String lockKey, boolean isSpin, int spinNum) {
		return lock(lockKey, isSpin, getSpinNum(redisConfig.getSpinNum(), DEFAULT_SPIN_NUM));
	}

	@Override
	public boolean lock(final String lockKey, final boolean isSpin, final int spinNum) {
		JedisCluster jedis = getJedis();
		try {
			if (log.isDebugEnabled()) {
				log.debug(
						"redis: method is acquire, key [" + lockKey + "],cache_redis [" + redisConfig.isEnable() + "]");
			}
			if (!redisConfig.isEnable()) {
				return true;
			}
			if (log.isDebugEnabled()) {
				log.debug("Start acquiring a lock  {},thread {}.", lockKey, Thread.currentThread().getId());
			}
			final String redisLockKey = getLockKey(lockKey);
			synchronized (redisLockKey) {
				long spinNumTemp = getSpinNum(redisConfig.getSpinNum(), DEFAULT_SPIN_NUM);
				while (spinNumTemp > 0) {
					if (log.isDebugEnabled()) {
						log.debug("start timeout:{},key:{} ", System.currentTimeMillis(), redisLockKey);
					}
					String expireTime = String.valueOf(System.currentTimeMillis() + redisConfig.getLockExpire());
					Long value = jedis.setnx(getLockKey(redisLockKey), expireTime);
					if (value >= 1) {
						if (log.isDebugEnabled()) {
							log.debug(" Manage to get a redis lock {},lockKey {} ", System.currentTimeMillis(),
									getLockKey(redisLockKey));
						}
						return true;
					} 
					
					if (log.isDebugEnabled()) {
						log.debug("lock timeout {},key {},thrread {} ", System.currentTimeMillis(), redisLockKey,
								Thread.currentThread().getId());
					}
					String currentValueStr = jedis.get(redisLockKey);
				    if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {
				        // 锁已过期，获取上一个锁的过期时间，并设置现在锁的过期时间
				        String oldValueStr = jedis.getSet(redisLockKey, expireTime);
				        if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {
				            // 考虑多线程并发的情况，只有一个线程的设置值和当前值相同，它才有权利加锁
				            return true;
				        }
				    }
					if (!isSpin) {// 如果不需要自旋获取锁，直接返回结果
						return false;
					}
					
	
					spinNumTemp -= 1;
					Thread.sleep(redisConfig.getSpinThreadTime());
				}
				}
				log.info("stop timeout:{},key:{},thread:{} ", System.currentTimeMillis(), redisLockKey,
						Thread.currentThread().getId());
				log.info("Failed to get the lock {},thread {} ", redisLockKey, Thread.currentThread().getId());
				
			return false;

		} catch (Exception e) {
			log.error(e.getMessage(), e);
			return true;
		}finally {
			returnResource(jedis);
		}
	}

	public void expire(JedisCluster jedis,String key, Long milliseconds) {
		jedis.pexpire(key, milliseconds);
	}
	
	   /**
     * 释放jedis资源
     * @param jedis
     */
    public void returnResource(final JedisCluster jedis) {
      /*  if (jedis != null) {
        	try {
				jedis.close();
			} catch (IOException e) {
				log.error(e.getMessage());
			}
        }*/
    }

	@Override
	public long incr(String key) {
		JedisCluster jedis = getJedis();
		try {
			return jedis.incr(key);
		} catch (Exception e) {
			log.error(e.getMessage());
		}finally {
			returnResource(jedis);
		}
		return 0;
	}

	@Override
	public Set<String> keys(String key) {
		return jedisCluster.hkeys(key);
	}

}

